package com.pw.study.flink.exercise;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;


public class W3 {

    public static void main(String[] args) {
        wc();
    }

    private static void wc() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(4);
        //输入数据
        DataStreamSource<String> data = env.readTextFile("data/file/words.txt");

        //mc1(data);
        mc2(data);

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void mc2(DataStreamSource<String> data) {
        data.flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        Arrays.stream(value.split(" ")).forEach(out::collect);
                    }
                }).map(x -> Tuple2.of(x, 1)).returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t -> t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Integer>, String>() {

                    private MapState<String, Integer> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Integer>("mapState", String.class, Integer.class));
                    }

                    @Override
                    public void processElement(Tuple2<String, Integer> value,Context ctx, Collector<String> out) throws Exception {

                        if(mapState.contains(value.f0)){
                            mapState.put(value.f0, mapState.get(value.f0)+1);
                        }else{
                            mapState.put(value.f0, value.f1);
                        }

                        System.out.println("====================");
                        for (String key : mapState.keys()) {
                            String msg="("+key+" : "+mapState.get(key)+")";
                            out.collect(msg);
                        }
                    }
                })
                .print();
    }

    private static void mc1(DataStreamSource<String> data) {
        data.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                Arrays.stream(value.split(" ")).forEach(x -> out.collect(Tuple2.of(x, 1L)));
            }
        }).keyBy(t -> t.f0).sum(1).print();
    }
}
